import traceback
from .db import tableDataFromJson, tableObjectFromJson, insertTable, createTable
from kafka import KafkaConsumer
from sqlalchemy import create_engine, MetaData
from json import loads, decoder


# 连接kafka broker
c = KafkaConsumer('app_log',  bootstrap_servers=[
                  '127.0.0.1:29092'], value_deserializer=lambda x: loads(x.decode('utf-8')))

# 连接postgresql
eng = create_engine(
    'postgresql+psycopg2://postgres:123456@127.0.0.1:25432/postgres')
conn = eng.connect()

metadata = MetaData()


def main():
    print('start...')
    tableDict = dict()
    # 反射获取当前所有表
    metadata.reflect(bind=eng)
    for key in metadata.tables.keys():
        tableDict[key] = metadata.tables[key]

    for msg in c:
        try:
            log = loads(msg.value['log'])
            print(log)
            if "type" not in log:
                continue
            tableName = log['type']
            tableData = tableDataFromJson(log)
            if(tableName not in tableDict):
                table = tableObjectFromJson(log, metadata)
                createTable(eng, metadata)
                tableDict[tableName] = table
            else:
                table = tableDict[tableName]
            insertTable(table, tableData, conn)

        except decoder.JSONDecodeError:
            print("json.decoder.JSONDecodeError")
            continue
        except Exception:
            print("Exception")
            traceback.print_exc()
            continue
